package com.game.core.db.service.jedis;

import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.Logger;

import com.game.core.db.common.util.Loggers;
import com.game.core.db.service.jedis.exceptions.JedisConnectionException;
import com.game.core.db.service.jedis.exceptions.JedisException;
import com.game.core.db.service.jedis.util.Pool;

public class JedisSentinelPool extends Pool<Jedis> {

	protected static Logger logger = Loggers.jedisLogger;
	protected GenericObjectPoolConfig poolConfig;

	protected int connectionTimeout = Protocol.DEFAULT_TIMEOUT;
	protected int soTimeout = Protocol.DEFAULT_TIMEOUT;

	protected String password;

	protected int database = Protocol.DEFAULT_DATABASE;

	protected String clientName;

	protected Set<MasterListener> masterListeners = new HashSet<MasterListener>();

	//protected Logger log = LoggerFactory.getLogger(getClass().getName());

	private volatile JedisFactory factory;
	private volatile HostAndPort currentHostMaster;

	public JedisSentinelPool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig) {
		this(masterName, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT, null, Protocol.DEFAULT_DATABASE);
	}

	public JedisSentinelPool(String masterName, Set<String> sentinels) {
		this(masterName, sentinels, new GenericObjectPoolConfig(), Protocol.DEFAULT_TIMEOUT, null,
				Protocol.DEFAULT_DATABASE);
	}

	public JedisSentinelPool(String masterName, Set<String> sentinels, String password) {
		this(masterName, sentinels, new GenericObjectPoolConfig(), Protocol.DEFAULT_TIMEOUT, password);
	}

	public JedisSentinelPool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig,
			int timeout, final String password) {
		this(masterName, sentinels, poolConfig, timeout, password, Protocol.DEFAULT_DATABASE);
	}

	public JedisSentinelPool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig,
			final int timeout) {
		this(masterName, sentinels, poolConfig, timeout, null, Protocol.DEFAULT_DATABASE);
	}

	public JedisSentinelPool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig,
			final String password) {
		this(masterName, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT, password);
	}

	public JedisSentinelPool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig,
			int timeout, final String password, final int database) {
		this(masterName, sentinels, poolConfig, timeout, timeout, password, database);
	}

	public JedisSentinelPool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig,
			int timeout, final String password, final int database, final String clientName) {
		this(masterName, sentinels, poolConfig, timeout, timeout, password, database, clientName);
	}

	public JedisSentinelPool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig,
			final int timeout, final int soTimeout, final String password, final int database) {
		this(masterName, sentinels, poolConfig, timeout, soTimeout, password, database, null);
	}

	public JedisSentinelPool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig,
			final int connectionTimeout, final int soTimeout, final String password, final int database,
			final String clientName) {
		this.poolConfig = poolConfig;
		this.connectionTimeout = connectionTimeout;
		this.soTimeout = soTimeout;
		this.password = password;
		this.database = database;
		this.clientName = clientName;

		HostAndPort master = initSentinels(sentinels, masterName);
		initPool(master);
	}

	@Override
	public void destroy() {
		for (MasterListener m : masterListeners) {
			m.shutdown();
		}

		super.destroy();
	}

	public HostAndPort getCurrentHostMaster() {
		return currentHostMaster;
	}

	private void initPool(HostAndPort master) {
		if (!master.equals(currentHostMaster)) {
			currentHostMaster = master;
			if (factory == null) {
				factory = new JedisFactory(master.getHost(), master.getPort(), connectionTimeout, soTimeout, password,
						database, clientName, false, null, null, null);
				initPool(poolConfig, factory);
			} else {
				factory.setHostAndPort(currentHostMaster);
				// although we clear the pool, we still have to check the
				// returned object
				// in getResource, this call only clears idle instances, not
				// borrowed instances
				internalPool.clear();
			}

			logger.info("Created JedisPool to master at " + master);
		}
	}

	private HostAndPort initSentinels(Set<String> sentinels, final String masterName) {

		HostAndPort master = null;
		boolean sentinelAvailable = false;

		logger.info("Trying to find master from available Sentinels...");

		for (String sentinel : sentinels) {
			final HostAndPort hap = HostAndPort.parseString(sentinel);

			logger.debug("Connecting to Sentinel {}", hap);

			Jedis jedis = null;
			try {
				jedis = new Jedis(hap.getHost(), hap.getPort());

				List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);

				// connected to sentinel...
				sentinelAvailable = true;

				if (masterAddr == null || masterAddr.size() != 2) {
					logger.warn("Can not get master addr, master name: {}. Sentinel: {}", masterName, hap);
					continue;
				}

				master = toHostAndPort(masterAddr);
				logger.debug("Found Redis master at {}", master);
				break;
			} catch (JedisException e) {
				// resolves #1036, it should handle JedisException there's another chance
				// of raising JedisDataException
				logger.warn("Cannot get master address from sentinel running @ {}. Reason: {}. Trying next one.", hap,
						e.toString());
			} finally {
				if (jedis != null) {
					jedis.close();
				}
			}
		}

		if (master == null) {
			if (sentinelAvailable) {
				// can connect to sentinel, but master name seems to not
				// monitored
				throw new JedisException(
						"Can connect to sentinel, but " + masterName + " seems to be not monitored...");
			} else {
				throw new JedisConnectionException(
						"All sentinels down, cannot determine where is " + masterName + " master is running...");
			}
		}

		logger.info("Redis master running at " + master + ", starting Sentinel listeners...");

		for (String sentinel : sentinels) {
			final HostAndPort hap = HostAndPort.parseString(sentinel);
			MasterListener masterListener = new MasterListener(masterName, hap.getHost(), hap.getPort());
			// whether MasterListener threads are alive or not, process can be stopped
			masterListener.setDaemon(true);
			masterListeners.add(masterListener);
			masterListener.start();
		}

		return master;
	}

	private HostAndPort toHostAndPort(List<String> getMasterAddrByNameResult) {
		String host = getMasterAddrByNameResult.get(0);
		int port = Integer.parseInt(getMasterAddrByNameResult.get(1));

		return new HostAndPort(host, port);
	}

	@Override
	public Jedis getResource() {
		while (true) {
			Jedis jedis = super.getResource();
			jedis.setDataSource(this);

			// get a reference because it can change concurrently
			final HostAndPort master = currentHostMaster;
			final HostAndPort connection = new HostAndPort(jedis.getClient().getHost(), jedis.getClient().getPort());

			if (master.equals(connection)) {
				// connected to the correct master
				return jedis;
			} else {
				returnBrokenResource(jedis);
			}
		}
	}

	/**
	 * @deprecated starting from Jedis 3.0 this method will not be exposed. Resource
	 *             cleanup should be done using @see
	 *             {@link com.game.core.db.service.jedis.Jedis#close()}
	 */
	@Override
	@Deprecated
	public void returnBrokenResource(final Jedis resource) {
		if (resource != null) {
			returnBrokenResourceObject(resource);
		}
	}

	/**
	 * @deprecated starting from Jedis 3.0 this method will not be exposed. Resource
	 *             cleanup should be done using @see
	 *             {@link com.game.core.db.service.jedis.Jedis#close()}
	 */
	@Override
	@Deprecated
	public void returnResource(final Jedis resource) {
		if (resource != null) {
			resource.resetState();
			returnResourceObject(resource);
		}
	}

	protected class MasterListener extends Thread {

		protected String masterName;
		protected String host;
		protected int port;
		protected long subscribeRetryWaitTimeMillis = 5000;
		protected volatile Jedis j;
		protected AtomicBoolean running = new AtomicBoolean(false);

		protected MasterListener() {
		}

		public MasterListener(String masterName, String host, int port) {
			super(String.format("MasterListener-%s-[%s:%d]", masterName, host, port));
			this.masterName = masterName;
			this.host = host;
			this.port = port;
		}

		public MasterListener(String masterName, String host, int port, long subscribeRetryWaitTimeMillis) {
			this(masterName, host, port);
			this.subscribeRetryWaitTimeMillis = subscribeRetryWaitTimeMillis;
		}

		@Override
		public void run() {

			running.set(true);

			while (running.get()) {

				j = new Jedis(host, port);

				try {
					// double check that it is not being shutdown
					if (!running.get()) {
						break;
					}

					j.subscribe(new JedisPubSub() {
						@Override
						public void onMessage(String channel, String message) {
							logger.debug("Sentinel {}:{} published: {}.", host, port, message);

							String[] switchMasterMsg = message.split(" ");

							if (switchMasterMsg.length > 3) {

								if (masterName.equals(switchMasterMsg[0])) {
									initPool(toHostAndPort(Arrays.asList(switchMasterMsg[3], switchMasterMsg[4])));
								} else {
									logger.debug(
											"Ignoring message on +switch-master for master name {}, our master name is {}",
											switchMasterMsg[0], masterName);
								}

							} else {
								logger.error("Invalid message received on Sentinel {}:{} on channel +switch-master: {}",
										host, port, message);
							}
						}
					}, "+switch-master");

				} catch (JedisException e) {

					if (running.get()) {
						logger.error("Lost connection to Sentinel at {}:{}. Sleeping 5000ms and retrying.", host, port, e);
						try {
							Thread.sleep(subscribeRetryWaitTimeMillis);
						} catch (InterruptedException e1) {
							logger.error("Sleep interrupted: InterruptedException err={}", e1.toString());
						}
					} else {
						logger.debug("Unsubscribing from Sentinel at {}:{}", host, port);
					}
				} finally {
					j.close();
				}
			}
		}

		public void shutdown() {
			try {
				logger.debug("Shutting down listener on {}:{}", host, port);
				running.set(false);
				// This isn't good, the Jedis object is not thread safe
				if (j != null) {
					j.disconnect();
				}
			} catch (Exception e) {
				logger.error("Caught exception while shutting down: Exception err={}", e.toString());
			}
		}
	}
}